package com.itheima.article.config;

import lombok.Data;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafkaStreams
@Data
@ConfigurationProperties(prefix = "kafka")
public class KafkaStreamConfig {

    private static final int MAX_MESSAGE_SIZE = 16*1024*1024;
    private String hosts;
    private String group;

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration defaultKafkaStreamsConfig(){
        Map<String,Object> props = new HashMap<>();

        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,hosts);

        props.put(StreamsConfig.APPLICATION_ID_CONFIG,this.getGroup() + "_stream_aid");
        props.put(StreamsConfig.CLIENT_ID_CONFIG,this.getGroup() + "_stream_aid");
        props.put(StreamsConfig.RETRIES_CONFIG,10);
        props.put(StreamsConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG,5000);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        return new KafkaStreamsConfiguration(props);
    }
}
